package cn.admin.remoting.impl;

import cn.admin.remoting.Connection;
import cn.admin.remoting.IOEventListener;
import cn.admin.remoting.RequestProcessor;
import cn.admin.remoting.locale.LogResources;
import cn.admin.remoting.util.DIYExecutor;
import cn.admin.remoting.util.LoggerInit;
import cn.admin.remoting.util.UnsafeCast;

import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;

class DefaultMsgListener implements IOEventListener.ConnectionMsgReceivedListener {

    private ProcessorUtil processors = new ProcessorUtil();

    @Override
    public void messageReceived(Connection conn, Object message) throws Exception {
        DefaultConnection connection = UnsafeCast.cast(conn);
        if (message instanceof ConnectionRequest) {
            LoggerInit.LOGGER.info("message coming = {}",message);
            this.doRequest(connection, (ConnectionRequest) message);
        } else {
            DefaultClient client = (DefaultClient) connection.getClient();
            client.putResponse((ConnectionResponse) message);
        }
    }

    @Override
    public void registerProcessor(RequestProcessor<?> processor) {
        this.processors.registerProcessor(processor);
    }

    @SuppressWarnings("unchecked")
    private void doRequest(final DefaultConnection connection, ConnectionRequest request) {
        final long requestId = request.getId();
        final byte direction = request.getDirection();
        final byte serializeProtocol = request.getSerializeProtocol();
        final Object appRequest = request.getAppRequest();
        final Thread ioThread = Thread.currentThread();
        connection.setLastRequestProtocol(serializeProtocol);
        final ConnectionResponse response = new ConnectionResponse();
        response.setRequestId(requestId);
        response.setDirection(direction);
        response.setSerializeProtocol(serializeProtocol);
        response.setHost(connection.getLocalAddress());
        final RequestProcessor.AppResponseOutput output = new AppResponseOutputImpl(connection,
                response);
        Class<?> appReqType = appRequest.getClass();
        final RequestProcessor processor = this.findProcessor(connection, appRequest);
        if (processor == null) {
            String msg = LogResources.getLog(LogResources.REQUEST_PROCESSOR_NOT_FOUND,
                    appReqType.getName());
            response.setResult(1);
            response.setErrorMsg(msg);
            connection.write(response, null);
        } else {
            final Executor requestExecutor =
                    processor instanceof RequestProcessor.MultiExecutorRequestProcessor ?
                            ((RequestProcessor.MultiExecutorRequestProcessor) processor).getExecutor(appRequest) :
                            processor.getExecutor();
            Runnable bisJob = () -> {
                if (ioThread == Thread.currentThread() && requestExecutor != DIYExecutor.getInstance()) {
                    String msg = LogResources.getLog(LogResources.IO_THREAD_CANNOT_DO_REQUEST,
                            requestId, appRequest);
                    response.setErrorMsg(msg);
                    connection.write(response, null);
                    LoggerInit.LOGGER.warn(msg);
                } else {
                    if (processor instanceof RequestProcessor.RequestPreProcessor) {
                        ((RequestProcessor.RequestPreProcessor) processor).
                                beforeHandleRequest(requestId, serializeProtocol, direction);
                    }

                    processor.handleRequest(appRequest, output);
                }
            };

            try {
                requestExecutor.execute(bisJob);
            } catch (RejectedExecutionException e) {
                processor.onRejectedExecutionException(appRequest, output);
            }
        }
    }

    private RequestProcessor<?> findProcessor(Connection conn, Object appRequest) {
        RequestProcessor<?> processor;
        processor = this.processors.findProcessor(appRequest.getClass());
        return null != processor ? processor :
                conn.getConnectionFactory().findProcessor(appRequest.getClass());
    }

    @Override
    public <T> RequestProcessor<T> removeProcessor(Class<T> appRequestClazz) {
        return this.processors.removeProcessor(appRequestClazz);
    }

    @Override
    public Map<Class<?>, RequestProcessor<?>> getProcessors() {
        return this.processors.getProcessors();
    }

    @Override
    public void updateProcessors(Map<Class<?>, RequestProcessor<?>> newProcessors) {
        this.processors.updateProcessors(newProcessors);
    }

    private static class AppResponseOutputImpl implements RequestProcessor.AppResponseOutput {

        private Connection connection;
        private ConnectionResponse connResp;
        private boolean isResponseSet = false;

        AppResponseOutputImpl(Connection connection, ConnectionResponse connResp) {
            this.connection = connection;
            this.connResp = connResp;
        }

        @Override
        public long requestId() {
            return this.connResp.getRequestId();
        }

        @Override
        public void write(Object appResp) {
            if (this.isResponseSet) {
                throw new IllegalStateException("########## 不能重复向AppResponseOutput写入响应.");
            } else {
                this.isResponseSet = true;
                if (this.connResp.getDirection() != 1) {
                    this.connResp.setAppResponse(appResp);
                    this.connection.write(this.connResp, null);
                }
            }
        }

        @Override
        public String remoteHost() {
            return this.connection.getRemoteAddress();
        }

        @Override
        public Connection getConnection() {
            return this.connection;
        }
    }

}
